package com.jy.oms.framework.asyn.task.checker;

import com.jy.oms.framework.asyn.task.cluster.config.ITaskConfigChangeListener;
import com.jy.oms.framework.asyn.task.constants.ConfigConstants;
import com.jy.oms.framework.asyn.task.constants.TimeUnitEnum;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import com.jy.oms.framework.asyn.task.model.strategy.LimiterExecuteStrategy;
import com.jy.oms.framework.util.SpringUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * 任务限流器配置监听器
 *
 * @author liuquan
 */
public class RateLimiterConfigListener implements ITaskConfigChangeListener {
    private String taskCode;
    private TaskHandlerWrapper taskHandlerWrapper;
    private Consumer<TaskHandlerWrapper> onConfigChangeCallback;

    public RateLimiterConfigListener(String taskCode, Consumer<TaskHandlerWrapper> onConfigChangeCallback) {
        this.taskCode = taskCode;

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        this.taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        this.onConfigChangeCallback = onConfigChangeCallback;
    }

    @Override
    public String getConfigKey() {
        return ConfigConstants.EXECUTOR_RATE_LIMITER + "_" + taskCode;
    }

    @Override
    public void onConfigChange(String configKey, String config) {
        if (StringUtils.isBlank(config)) {
            return;
        }

        this.changeExecuteStrategyConfig(config);

        if (onConfigChangeCallback != null) {
            onConfigChangeCallback.accept(taskHandlerWrapper);
        }
    }

    @Override
    public void onConfigChangeAfterLoaded(String configKey, String config) {
        if(StringUtils.isNotBlank(config)){
            this.onConfigChange(configKey, config);
        }else{
            if (onConfigChangeCallback != null) {
                onConfigChangeCallback.accept(taskHandlerWrapper);
            }
        }
    }

    private void changeExecuteStrategyConfig(String config) {
        Map<String, String> configMap = new HashMap<>();
        Arrays.stream(config.split(",")).forEach(c -> {
            String[] values = c.split(":");
            if (values.length == 2) {
                configMap.put(StringUtils.trimToEmpty(values[0]), StringUtils.trimToEmpty(values[1]));
            }
        });
        TimeUnitEnum timeUnit = convertTimeUnitEnum(configMap.get("timeUnit"));
        if (timeUnit == null) {
            return;
        }
        int interval = NumberUtils.toInt(configMap.get("interval"));
        if (interval == 0) {
            return;
        }
        int limiter = NumberUtils.toInt(configMap.get("limiter"));
        if (limiter == 0) {
            return;
        }

        LimiterExecuteStrategy executeStrategy = (LimiterExecuteStrategy) taskHandlerWrapper.getExecuteStrategy();
        executeStrategy.setTimeUnit(timeUnit);
        executeStrategy.setInterval(interval);
        executeStrategy.setLimiter(limiter);
    }

    private TimeUnitEnum convertTimeUnitEnum(String timeUnit) {
        if (StringUtils.isBlank(timeUnit)) {
            return null;
        }
        if (StringUtils.equalsIgnoreCase(timeUnit, "second")) {
            return TimeUnitEnum.Second;
        } else if (StringUtils.equalsIgnoreCase(timeUnit, "minute")) {
            return TimeUnitEnum.Minute;
        } else if (StringUtils.equalsIgnoreCase(timeUnit, "hour")) {
            return TimeUnitEnum.Hour;
        } else if (StringUtils.equalsIgnoreCase(timeUnit, "day")) {
            return TimeUnitEnum.Day;
        }

        return null;
    }
}
